package io.kiki.sba.registry.server.data_store.multi.cluster.storage;


import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.RegisterVersion;
import io.kiki.sba.registry.common.model.dataserver.Datum;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.slot.SlotRole;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.exception.UnSupportOperationException;
import io.kiki.sba.registry.server.data_store.cache.CleanContinues;
import io.kiki.sba.registry.server.data_store.cache.IDatumStorage;
import io.kiki.sba.registry.server.data_store.cache.PublisherGroup;
import io.kiki.sba.registry.server.data_store.cache.SlotIdToPublisherGroupsMap;
import io.kiki.sba.registry.server.data_store.slot.SlotChangeListener;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;


public class MultiClusterDatumStorage implements IDatumStorage {

    private static final Logger logger = LoggerFactory.getLogger(MultiClusterDatumStorage.class);

    // map<dataCenter, storage>
    private final Map<String, SlotIdToPublisherGroupsMap> storageMap = new ConcurrentHashMap<>();

    private final MultiClusterSlotListener multiClusterSlotListener = new MultiClusterSlotListener();

    @Override
    public Set<String> allDataCenters() {
        return storageMap.keySet();
    }


    @Override
    public Datum get(String dataCenter, String dataInfoId) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]dataCenter={},dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        Datum datum = slotIdToPublisherGroupsMap.get(dataInfoId);
        if (datum == null) {
            logger.warn("[nullDatum]dataCenter={},dataInfoId={}", dataCenter, dataInfoId);
        }
        return datum;
    }

    @Override
    public DatumVersion getVersion(String dataCenter, String dataInfoId) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]getVersion dataCenter={},dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        return slotIdToPublisherGroupsMap.getVersion(dataInfoId);
    }

    @Override
    public Map<String, DatumVersion> getVersions(String dataCenter, int slotId, Collection<String> targetDatInfoIds) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]getVersions dataCenter={},slotId={}, targetDatInfoIds={}", dataCenter, slotId, targetDatInfoIds);
            return Collections.emptyMap();
        }
        return slotIdToPublisherGroupsMap.getVersions(slotId, targetDatInfoIds);
    }

    @Override
    public Map<String, Publisher> getByConnectId(ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId) {
        throw new UnSupportOperationException("MultiClusterDatumStorage.getByConnectId");
    }

    @Override
    public Map<String, Map<String, Publisher>> getDataInfoIdToRegisterIdToPublisherMapMap(String dataCenter, int slotId) {

        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]getPublishers dataCenter={},slotId={}", dataCenter, slotId);
            return Collections.emptyMap();
        }
        return slotIdToPublisherGroupsMap.getDataInfoIdToRegisterIdToPublisherMapMap(slotId);
    }


    @Override
    public Map<String, Datum> getDataInfoIdToDatumMap(String dataCenter) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]getAll dataCenter={}", dataCenter);
            return Collections.emptyMap();
        }
        return slotIdToPublisherGroupsMap.getDataInfoIdToDatumMap();
    }

    @Override
    public Map<String, List<Publisher>> getDataInfoIdToPublisherListMap(String dataCenter) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]getAllPublisher dataCenter={}", dataCenter);
            return Collections.emptyMap();
        }
        return slotIdToPublisherGroupsMap.getDataInfoIdToPublisherListMap();
    }

    @Override
    public Map<String, Integer> getPubCount(String dataCenter) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.get(dataCenter);
        if (slotIdToPublisherGroupsMap == null) {
            logger.warn("[nullStorage]getPubCount dataCenter={}", dataCenter);
            return Collections.emptyMap();
        }
        return slotIdToPublisherGroupsMap.getPubCount();
    }

    @Override
    public void putPublisherGroups(String dataCenter, int slotId) {
        SlotIdToPublisherGroupsMap slotIdToPublisherGroupsMap = storageMap.computeIfAbsent(dataCenter, k -> new SlotIdToPublisherGroupsMap(dataCenter));
        slotIdToPublisherGroupsMap.putPublisherGroups(slotId);
    }

    @Override
    public DatumVersion putPublisher(String dataCenter, Publisher publisher) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]putPublisher dataCenter={}, dataInfoId={}, registerId={}", dataCenter, publisher.getDataInfoId(), publisher.getRegisterId());
            return null;
        }
        return storage.putPublisher(publisher);
    }

    @Override
    public DatumVersion createEmptyDatumIfAbsent(String dataCenter, String dataInfoId) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]createEmptyDatumIfAbsent dataCenter={}, dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        return storage.createEmptyDatumIfAbsent(dataInfoId);
    }

    @Override
    public Map<String, DatumVersion> cleanByClientInterfaceId(String dataCenter, int slotId, ClientInterfaceId clientInterfaceId, CleanContinues cleanContinues) {
        logger.error("[MultiClusterDatumStorage]UnExcept cleanBySessionId, dataCenter={}, slotId={}, sessionProcessId={}", dataCenter, slotId, clientInterfaceId);
        throw new UnSupportOperationException("MultiClusterDatumStorage.cleanBySessionId");
    }

    @Override
    public boolean removePublisherGroups(String dataCenter, int slotId) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]removePublisherGroups dataCenter={}, slotId={}", dataCenter, slotId);
            return false;
        }
        return storage.removePublisherGroups(slotId);
    }

    @Override
    public DatumVersion removePublishers(String dataCenter, String dataInfoId, ClientInterfaceId clientInterfaceId) {
        logger.error("[MultiClusterDatumStorage]UnExcept removePublishers, dataCenter={}, dataInfoId={}, sessionProcessId={}", dataCenter, dataInfoId, clientInterfaceId);
        throw new UnSupportOperationException("MultiClusterDatumStorage.removePublishersBySessionId");
    }

    @Override
    public DatumVersion removePublishers(String dataCenter, ClientInterfaceId clientInterfaceId, String dataInfoId, Map<String, RegisterVersion> removedPublishers) {
        StringUtil.nonNullValue(clientInterfaceId, "sessionProcessId");

        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]removePublishers dataCenter={}, dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        return storage.removePublishers(dataInfoId, null, removedPublishers);
    }

    @Override
    public DatumVersion putPublisher(String dataCenter, String dataInfoId, List<Publisher> updatedPublishers) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]putPublisher dataCenter={}, dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        return storage.putPublisher(dataInfoId, updatedPublishers);
    }

    @Override
    public SlotChangeListener getSlotChangeListener(boolean localDataCenter) {
        return multiClusterSlotListener;
    }

    @Override
    public Set<ClientInterfaceId> getClientInterfaceIdSet(String dataCenter) {
        logger.error("[MultiClusterDatumStorage]UnExcept getSessionProcessIds, dataCenter={}", dataCenter);
        throw new UnSupportOperationException("MultiClusterDatumStorage.getSessionProcessIds");
    }

    @Override
    public Map<String, Integer> compact(String dataCenter, long tombstoneTimestamp) {
        logger.error("[MultiClusterDatumStorage]UnExcept getSessionProcessIds, dataCenter={}", dataCenter);
        throw new UnSupportOperationException("MultiClusterDatumStorage.getSessionProcessIds");
    }

    @Override
    public int tombstoneNum(String dataCenter) {
        logger.error("[MultiClusterDatumStorage]UnExcept tombstoneNum, dataCenter={}", dataCenter);
        throw new UnSupportOperationException("MultiClusterDatumStorage.tombstoneNum");
    }

    @Override
    public Map<String, DatumVersion> updateVersion(String dataCenter, int slotId) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]updateVersion dataCenter={}, slotId={}", dataCenter, slotId);
            return Collections.emptyMap();
        }
        return storage.updateVersion(slotId);
    }

    @Override
    public DatumVersion updateVersion(String dataCenter, String dataInfoId) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]updateVersion dataCenter={}, dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        return storage.updateVersion(dataInfoId);
    }

    @Override
    public void foreach(String dataCenter, int slotId, BiConsumer<String, PublisherGroup> f) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]foreach dataCenter={}, slotId={}", dataCenter, slotId);
            return;
        }
        storage.foreach(slotId, f);
    }

    @Override
    public boolean removeStorage(String dataCenter) {
        storageMap.remove(dataCenter);
        return true;
    }

    @Override
    public DatumVersion clearPublishers(String dataCenter, String dataInfoId) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]clearPublishers dataCenter={}, dataInfoId={}", dataCenter, dataInfoId);
            return null;
        }
        return storage.clearPublishers(dataInfoId);
    }

    @Override
    public Map<String, DatumVersion> clearGroupPublishers(String dataCenter, String group) {
        SlotIdToPublisherGroupsMap storage = storageMap.get(dataCenter);
        if (storage == null) {
            logger.warn("[nullStorage]clearGroupPublishers dataCenter={}, group={}", dataCenter, group);
            return null;
        }
        return storage.clearGroupPublishers(group);
    }

    private final class MultiClusterSlotListener implements SlotChangeListener {

        @Override
        public void onSlotAdd(String dataCenter, int slotId, SlotRole slotRole) {
            putPublisherGroups(dataCenter, slotId);
            logger.info("{} add publisherGroup {}, role={},", dataCenter, slotId, slotRole);
        }

        @Override
        public void onSlotRemove(String dataCenter, int slotId, SlotRole slotRole) {
            boolean removed = removePublisherGroups(dataCenter, slotId);
            logger.info("{}, remove publisherGroup {}, removed={}, role={}", dataCenter, slotId, removed, slotRole);
        }
    }
}
